package cn.com.cmbc;

import cn.com.cmbc.config.FileConfig;
import cn.com.cmbc.dao.HBaseDAO;
import cn.com.cmbc.dao.KafkaDAO;
import cn.com.cmbc.dto.HbaseDTO;
import cn.com.cmbc.service.HbaseService;
import cn.com.cmbc.service.KafkaService;
import cn.com.cmbc.utils.HTTPUtil;
import cn.com.cmbc.utils.JSONUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Hello world!
 *
 */
public class Json2Hbase {
    private static final Logger LOG = LoggerFactory.getLogger(Json2Hbase.class);
    public static void main( String[] args ) throws IOException {
        System.out.println( "Hello World!" );
        String url = "http://localhost:8080/getfiles?path="+ FileConfig.FILE_PATH;

        String json = HTTPUtil.get(url);

        String fileList = JSONUtil.getMessage(json);

        HbaseService hbaseService = new HbaseService();
        List<HbaseDTO> hbaseDTOS = hbaseService.json2DTO(fileList);

//        hbaseDTOS.forEach(System.out::println);
        List<HBaseDAO> hBaseDAOS = new ArrayList<>();

        hbaseDTOS.forEach(hbaseDTO -> {
            HBaseDAO hBaseDAO = new HBaseDAO(hbaseDTO.getName(), hbaseDTO.getLastModifiedTime(), hbaseDTO.getType(), hbaseDTO.getSuffix(), hbaseDTO.getSize(), hbaseDTO.getAbsolutePath(), hbaseDTO.getRowkeyId());
            hBaseDAOS.add(hBaseDAO);
        });


        LOG.info("{}条数据正在存入Hbase。。。。。。。", hBaseDAOS.size());

        String tableName = "bigdata:diskfile";
        String family = "base_info";

        boolean putResult = hbaseService.putListData(tableName, family, hBaseDAOS);
        if (putResult){
            LOG.info("数据成功存入HBase");
        }else {
            LOG.info("数据存入HBase失败");
            return;
        }


        //数据存入kafka
        List<KafkaDAO> kafkaDAOS = new ArrayList<>();

        hbaseDTOS.forEach(hbaseDTO -> {
            KafkaDAO kafkaDAO = new KafkaDAO(hbaseDTO.getName(), hbaseDTO.getLastModifiedTime(), hbaseDTO.getType(), hbaseDTO.getSuffix(), hbaseDTO.getSize(), hbaseDTO.getAbsolutePath(), hbaseDTO.getRowkeyId());
            kafkaDAOS.add(kafkaDAO);
        });

        String topic = "diskfile";

        KafkaService kafkaService = new KafkaService();
        boolean status = kafkaService.sendKafkaList(topic, kafkaDAOS);

        if (status){
            LOG.info("数据成功发送Kafka");
        }else {
            LOG.info("数据发送Kafka失败");
            return;
        }

    }
}
